// Copyright (C) Kumo inc. and its affiliates.
// Author: Jeff.li lijippy@163.com
// All rights reserved.
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published
// by the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// This program is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
// GNU Affero General Public License for more details.
//
// You should have received a copy of the GNU Affero General Public License
// along with this program.  If not, see <https://www.gnu.org/licenses/>.
//

#include <nebula/filesystem/filesystem.h>
#include <nebula/filesystem/filesystem_library.h>
#include <nebula/io/memory.h>

#include <turbo/uri/uri.h>

// Demonstrate registering a user-defined Nebula FileSystem outside
// of the Nebula source tree.

namespace io = nebula::io;
namespace fs = nebula::fs;

class ExampleFileSystem : public fs::FileSystem {
 public:
  explicit ExampleFileSystem(const io::IOContext& io_context)
      : fs::FileSystem{io_context} {}

  // This is a mock filesystem whose root directory contains a single file.
  // All operations which would mutate will simply raise an error.
  static constexpr std::string_view kPath = "example_file";
  static constexpr std::string_view kContents = "hello world";
  static fs::FileInfo info() {
    fs::FileInfo info;
    info.set_path(std::string{kPath});
    info.set_type(fs::FileType::File);
    info.set_size(kContents.size());
    return info;
  }

  static turbo::Status NotFound(std::string_view path) {
    return turbo::io_error("Path does not exist '", path, "'");
  }

  static turbo::Status NoMutation() {
    return turbo::io_error("operations which would mutate are not permitted");
  }

  turbo::Result<std::string> PathFromUri(const std::string& uri_string) const override {
    TURBO_MOVE_OR_RAISE(auto uri, turbo::Uri::from_string(uri_string));
    return uri.path();
  }

  std::string type_name() const override { return "example"; }

  bool equals(const FileSystem& other) const override {
    return type_name() == other.type_name();
  }

  /// \cond FALSE
  using FileSystem::CreateDir;
  using FileSystem::DeleteDirContents;
  using FileSystem::GetFileInfo;
  using FileSystem::open_append_stream;
  using FileSystem::open_output_stream;
  /// \endcond

  turbo::Result<fs::FileInfo> GetFileInfo(const std::string& path) override {
    if (path == kPath) {
      return info();
    }
    return NotFound(path);
  }

  turbo::Result<std::vector<fs::FileInfo>> GetFileInfo(const fs::FileSelector& select) override {
    if (select.base_dir == "/" || select.base_dir == "") {
      return std::vector<fs::FileInfo>{info()};
    }
    if (select.allow_not_found) {
      return std::vector<fs::FileInfo>{};
    }
    return NotFound(select.base_dir);
  }

  turbo::Status CreateDir(const std::string& path, bool recursive) override {
    return NoMutation();
  }

  turbo::Status DeleteDir(const std::string& path) override { return NoMutation(); }

  turbo::Status DeleteDirContents(const std::string& path, bool missing_dir_ok) override {
    return NoMutation();
  }

  turbo::Status DeleteRootDirContents() override { return NoMutation(); }

  turbo::Status DeleteFile(const std::string& path) override { return NoMutation(); }

  turbo::Status Move(const std::string& src, const std::string& dest) override {
    return NoMutation();
  }

  turbo::Status CopyFile(const std::string& src, const std::string& dest) override {
    return NoMutation();
  }

  turbo::Result<std::shared_ptr<io::InputStream>> open_input_stream(
      const std::string& path) override {
    return OpenInputFile(path);
  }

  turbo::Result<std::shared_ptr<io::RandomAccessFile>> OpenInputFile(
      const std::string& path) override {
    if (path == kPath) {
      return io::BufferReader::from_string(std::string{kContents});
    }
    return NotFound(path);
  }

  turbo::Result<std::shared_ptr<io::OutputStream>> open_output_stream(
      const std::string& path,
      const std::shared_ptr<const nebula::KeyValueMetadata>& metadata) override {
    return NoMutation();
  }

  turbo::Result<std::shared_ptr<io::OutputStream>> open_append_stream(
      const std::string& path,
      const std::shared_ptr<const nebula::KeyValueMetadata>& metadata) override {
    return NoMutation();
  }
};

auto kExampleFileSystemModule = NEBULA_REGISTER_FILESYSTEM(
    "example",
    [](const turbo::Uri& uri, const io::IOContext& io_context,
       std::string* out_path) -> turbo::Result<std::shared_ptr<fs::FileSystem>> {
      auto fs = std::make_shared<ExampleFileSystem>(io_context);
      if (out_path) {
        TURBO_MOVE_OR_RAISE(*out_path, fs->PathFromUri(uri.to_string()));
      }
      return fs;
    },
    {});
